package com.system.mqConsumer;

import java.util.concurrent.atomic.AtomicLong;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.lg.service.RetryService;
import com.system.constants.QueueConstant;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class ConsumerListener {

	@Autowired
	private RetryService kafkaRetryService;

	private static AtomicLong atomicLong = new AtomicLong();

//	@KafkaListener(topics = QueueConstant.PROC_TEST_TOPIC, groupId = QueueConstant.LG_DEFAULT_GROUP)
	public void consume(ConsumerRecord<String, String> record) {
		
		try {
			// 业务处理
			log.info("消费的消息：{}", record);
			long decrementAndGet = atomicLong.incrementAndGet();
			if (decrementAndGet % 2 == 0) {
				throw new Exception("该重发了");
			}
		} catch (Exception e) {
			log.error(e.getMessage());
			// 消息重试，实际上先将消息放到redis
			kafkaRetryService.consumerLater(record);
		}
	}

}